package com.frinder.hadoop.hbase.original;

import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.Result;

import java.util.List;

/**
 * @author frinder
 * @date 2018/6/30
 * @Description: 代理类，管理连接，获取 & 归还连接
 */
public class HBaseHelperProxy {

    protected static ConnectionProvider provider = new MyConnectionProvider();

    private HBaseHelper helper;

    public HBaseHelperProxy() {
        helper = new HBaseHelper();
    }

    public Connection getConn() throws Exception {
        return provider.getConn();
    }

    /**
     * 创建表
     *
     * @param tableName
     * @param cols
     * @throws Exception
     */
    public void createTable(String tableName, String... cols) throws Exception {
        Connection conn = getConn();
        try {
            helper.createTable(conn, tableName, cols);
        } finally {
            provider.returnConn(conn);
        }
    }

    /**
     * 删除表
     *
     * @param tableName
     * @throws Exception
     */
    public void deleteTable(String tableName) throws Exception {
        Connection conn = getConn();
        try {
            helper.deleteTable(conn, tableName);
        } finally {
            provider.returnConn(conn);
        }
    }

    /**
     * 获取表
     *
     * @return
     * @throws Exception
     */
    public List<TableName> getTables() throws Exception {
        Connection conn = getConn();
        try {
            return helper.getTables(conn);
        } finally {
            provider.returnConn(conn);
        }
    }

    /**
     * 插入记录
     *
     * @param tableName
     * @param rowKey
     * @param colFamily
     * @param col
     * @param val
     * @throws Exception
     */
    public void insertRow(String tableName, String rowKey, String colFamily, String col, String val) throws Exception {
        Connection conn = getConn();
        try {
            helper.insertRow(conn, tableName, rowKey, colFamily, col, val);
        } finally {
            provider.returnConn(conn);
        }
    }

    /**
     * 删除记录
     *
     * @param tableName
     * @param rowKey
     * @throws Exception
     */
    public void deleteRow(String tableName, String rowKey) throws Exception {
        Connection conn = getConn();
        try {
            helper.deleteRow(conn, tableName, rowKey);
        } finally {
            provider.returnConn(conn);
        }
    }

    /**
     * 删除 簇列
     *
     * @param tableName
     * @param rowKey
     * @param colFamily
     * @param col
     * @throws Exception
     */
    public void deleteCol(String tableName, String rowKey, String colFamily, String col) throws Exception {
        Connection conn = getConn();
        try {
            helper.deleteCol(conn, tableName, rowKey, colFamily, col);
        } finally {
            provider.returnConn(conn);
        }
    }

    /**
     * 获取记录
     *
     * @param tableName
     * @param rowKey
     * @return
     * @throws Exception
     */
    public Result getRowData(String tableName, String rowKey) throws Exception {
        Connection conn = getConn();
        try {
            return helper.getRowData(conn, tableName, rowKey);
        } finally {
            provider.returnConn(conn);
        }
    }

    /**
     * 获取 簇列
     *
     * @param tableName
     * @param rowKey
     * @param colFamily
     * @param col
     * @return
     * @throws Exception
     */
    public Result getColData(String tableName, String rowKey, String colFamily, String col) throws Exception {
        Connection conn = getConn();
        try {
            return helper.getColData(conn, tableName, rowKey, colFamily, col);
        } finally {
            provider.returnConn(conn);
        }
    }

    /**
     * 批量查询记录
     *
     * @param tableName
     * @param startRow
     * @param stopRow
     * @return
     * @throws Exception
     */
    public List<Result> scanData(String tableName, String startRow, String stopRow) throws Exception {
        Connection conn = getConn();
        try {
            return helper.scanData(conn, tableName, startRow, stopRow);
        } finally {
            provider.returnConn(conn);
        }
    }
}
